package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

/**
 * spark streaming中分为两种窗口：
 * 窗口是由两部分构成【1.窗口大小  2.滑动的大小】
 *
 * 1、滚动窗口：窗口的大小和滑动的大小是一样
 * 2、滑动窗口：窗口的大小和滑动不一样，前后批次的数据会被两个不同的job作业进行处理
 */
object WindowDemo {
  def main(args: Array[String]): Unit = {
    /**
     * Spark core:        SparkContext -> RDD
     * Spark Sql:         SparkSession -> DataFrame
     * Spark Streaming:   StreamingContext -> DStream
     */
    //this(sparkContext: SparkContext, batchDuration: Duration)

    val ss: SparkSession = SparkSession.builder()
      .master("local[*]") // 如果不给线程核数，默认只有1个
      .appName("word count")
      .getOrCreate()

    val sc: SparkContext = ss.sparkContext
    sc.setCheckpointDir("spark/data/checkpoint2")

    //    val conf = new SparkConf()
    //    conf.setMaster("local")
    //    conf.setAppName("word count")
    //    val sc = new SparkContext(conf)

    //传入一个SparkContext对象，以及接收要处理的数据时间段
    val streamingContext = new StreamingContext(sc, Durations.seconds(5L))

    //近实时处理中，数据应当是源源不断的
    //使用端口号的数据，模拟源源不断的数据源
    //监控对应端口号中的数据
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)
    //不同的DStream可以单独的设置窗口和滑动步长
//    infoDS.window(Durations.seconds(10),Durations.seconds(20))



    // hello world hello java
    //因为DStream中封装的是RDD,所以可以使用RDD中的大部分的算子
    val wordsDS: DStream[String] = infoDS.flatMap(_.split(" "))
    val wordKVDS: DStream[(String, Int)] = wordsDS.map((_, 1))
    //    val resDS: DStream[(String, Int)] = wordKVDS.reduceByKey(_ + _)
    /**
     * reduceByKeyAndWindow(
     * reduceFunc: (V, V) => V,    针对前一个数据和后一个数据的处理逻辑
     * windowDuration: Duration,   窗口的大小
     * slideDuration: Duration     滑动的步长
     * )
     *
     * 每次滑动产生一个job作业执行！！！
     *
     * 注意：
     *  这里设置的窗口大小和滑动步长必须是上面数据源设置窗口大小的整数倍
     */
    val resDS: DStream[(String, Int)] = wordKVDS.reduceByKeyAndWindow((a: Int, b: Int) => a + b, Durations.seconds(5L), Durations.seconds(10L))


    println("-----------------------------------")
    resDS.print()
    println("-----------------------------------")

    /**
     * 额外提交命令，一直启动sparkstreaming程序
     */
    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }

}
